package com.we.risk.registerchanneldeprecated.basicinfo.dim;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import com.alibaba.fastjson.JSONObject;

public class TableUserPhoneInfoHive {
    public static final String KFKTABLE = "adm.rp_user_phone_info_all";
    public static final String QUERYSQL =
            "select uid," + " idfa as id_combine" + " from " + KFKTABLE;
    public long uid;
    public String id_combine;

    public TableUserPhoneInfoHive() {}

    public TableUserPhoneInfoHive(long uid, String id_combine) {
        this.uid = uid;
        this.id_combine = id_combine;
    }

    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        tableEnv.useCatalog("cdh2_hive");
        TableConfig tableConf = tableEnv.getConfig();
        Configuration selfConf = new Configuration();
        selfConf.setBoolean("table.exec.hive.infer-source-parallelism", false);
        tableConf.addConfiguration(selfConf);
        Table tableRes = tableEnv.sqlQuery(TableUserPhoneInfoHive.QUERYSQL);
        DataStream<TableUserPhoneInfoHive> inputStream =
                tableEnv.toDataStream(tableRes, TableUserPhoneInfoHive.class);

        SingleOutputStreamOperator<String> out =
                inputStream.map(
                        new RichMapFunction<TableUserPhoneInfoHive, String>() {
                            @Override
                            public String map(TableUserPhoneInfoHive value) throws Exception {
                                JSONObject outJson = new JSONObject();
                                outJson.put("uid", value.getUid());
                                outJson.put("id_combine", value.getId_combine());

                                return outJson.toString();
                            }
                        });
        return out;
    }

    public long getUid() {
        return uid;
    }

    public String getId_combine() {
        return id_combine;
    }
}
